package cn.demo.io.demo3.training;

import java.io.*;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 服务端应用程序
 *
 * @author admin
 */
public class Server {
    /**
     * 服务端Socket
     */
    private ServerSocket server;

    /**
     * 线程池
     * 用于指派接受客户端数据的任务
     */
    private ExecutorService threadPool;

    /**
     * 存放所有客户端输出流的集合
     * 用于广播
     */
    private List<PrintWriter> allOut;

    /**
     * 创建一个消息队列
     * 用于存放所有用户发送过来的消息
     * java.util.concurrent.BlockingDeque
     */
    private BlockingDeque<String> messageQueue;


    /**
     * 构造方法，用于初始化
     */
    public Server() {
        try {
            /*
			 * 创建ServerSocket，并申请服务端口
			 */
            System.out.println("初始化服务端口...");
            server = new ServerSocket(8088);
            System.out.println("端口初始化完毕!");
			/*
			 * 初始化线程池
			 */
            threadPool = Executors.newCachedThreadPool();

			/*
			 * 初始化存放所有客户端输出流的集合
			 */
            allOut = new ArrayList<PrintWriter>();

			/*
			 * 初始化消息队列
			 */
            messageQueue = new LinkedBlockingDeque<String>();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 启动服务端的方法
     */
    public void start() {
        try {
			/*
			 * 启动服务端时，先将转发客户端消息的线程
			 * 启动起来
			 */
            Runnable sendMessageHandler = new SendMessageToAllClientHandler();
            Thread t = new Thread(sendMessageHandler);
            t.setDaemon(true);
            t.start();

			/*
			 * Socket accept()
			 * 该方法用于监听8088端口，等待客户端的链接
			 * 该方法是一个阻塞方法，调用该方法后直到
			 * 一个客户端链接，accept()方法才会将该客户端
			 * Socket返回，否则一直处于阻塞。
			 *
			 * Socket套接字
			 */
            while (true) {
                System.out.println("等待客户端链接...");
                Socket socket = server.accept();
                System.out.println("一个客户端链接了!");
				/*
				 * 创建一个任务，将刚刚链接上得客户端Socket
				 * 传入
				 */
                Runnable handler = new GetClientMessageHandler(socket);
                //启动一个线程去处理该用户发送的数据
//				Thread t = new Thread(handler);
//				t.start();

				/*
				 * 不再使用上面自己创建线程的方式
				 * 而是将任务Runnable指派到线程池
				 */
                threadPool.execute(handler);
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 将给定的输出流存入共享集合
     * synchronized修饰方法:
     * 被修饰的方法为同步方法，若一个类中多个方法
     * 被synchronized修饰，那么这些方法是互斥方法
     * 即:一个线程在访问其中一个方法时，其他线程对这些
     * 方法均不能访问
     *
     * @param out
     */
    private synchronized void addOut(PrintWriter out) {
        allOut.add(out);
    }

    /**
     * 将给定的输出流从共享集合中删除
     *
     * @param out
     */
    private synchronized void removeOut(PrintWriter out) {
        allOut.remove(out);
    }

    /**
     * 将给定的消息发送给所有的客户端
     * <p>
     * 该方法需要与上面两个方法互斥，原因在于集合迭代时
     * 有一个要求，就是迭代过程中不能通过集合对元素进行
     * 增删操作。否则迭代会出现异常。
     *
     * @param message
     */
    private synchronized void sendMessageToAllClient(String message) {
		/*
		 * 遍历所有的输出流，将给定的字符串写出即可
		 */
        for (PrintWriter writer : allOut) {
            writer.println(message);
        }
    }


    /**
     * 定义线程要执行的任务
     * 每一个线程用于和一个客户端交流
     * 读取客户端发送过来的数据并输出到控制台
     *
     * @author admin
     */
    private class GetClientMessageHandler implements Runnable {
        //该线程需要交流的客户端Socket
        private Socket socket;
        //客户端的地址
        private String clientHost;

        public GetClientMessageHandler(Socket socket) {
            this.socket = socket;
            //java.net.InetAddress
			/*
			 * InetAddress getInetAddress()
			 * 获取该客户端的相关地址信息
			 */
            /*InetAddress clientAddress = socket.getInetAddress();
            //获取客户端的IP
            clientHost = clientAddress.getHostAddress();
            System.out.println(clientHost + "上线了.");*/

            SocketAddress remoteSocketAddress = socket.getRemoteSocketAddress();
            System.out.println(remoteSocketAddress + "上线了.");
        }

        public void run() {
            PrintWriter writer = null;//该客户端的输出流
            try {
				/*
				 * 创建对该客户端的输出流，并存入共享的
				 * 集合中。
				 */
                OutputStream out = socket.getOutputStream();
                OutputStreamWriter osw = new OutputStreamWriter(out, "UTF-8");
                writer = new PrintWriter(osw, true);

                //将该客户端的输出流存入共享集合
//				allOut.add(writer); 不在这里存
                addOut(writer);
                System.out.println("当前在线人数:" + allOut.size());

				/*
				 * 获取该客户端的输入流，循环读取该客户端
				 * 发送过来的信息，并输出到控制台
				 */
				/*
				 * 通过该客户端的Socket获取输入流
				 * 用于读取客户端发送过来的数据
				 */
                InputStream in = socket.getInputStream();
                InputStreamReader isr = new InputStreamReader(in, "UTF-8");
                BufferedReader reader = new BufferedReader(isr);

                //读取客户端发送过来的一行字符串
                String message = null;
                while ((message = reader.readLine()) != null) {
//					System.out.println(clientHost+"说:"+message);
//					sendMessageToAllClient(clientHost+"说:"+message);
					/*
					 * 不直接做转发，存在多线程向同一个输出流写操作的
					 * 线程安全问题，我们将信息存入消息队列，等待统一
					 * 的方式转发
					 */
                    messageQueue.offer(clientHost + "说:" + message);
					/*
					 * 双缓冲队列在添加元素时允许设置超时时间
					 * 当队列元素已满时，可以等待，若超时时间
					 * 到了还没有被加入队列才返回false
					 */
//					messageQueue.offer(
//						clientHost+"说:"+message,
//						5,
//						TimeUnit.SECONDS
//						);
                }

            } catch (Exception e) {
            } finally {
				/*
				 * 当客户端关闭程序(与服务端断开链接)后
				 * 若是linux的客户端，
				 * 那么服务端message = reader.readLine()
				 * 会读取到null
				 *
				 * 若使windows得客户端
				 * message = reader.readLine()会抛异常
				 *
				 * 总之，客户端与服务端断开连接后，那么读取
				 * 该客户端数据的当前线程就应当结束了
				 * 随之也应当将这个客户端的链接关闭，释放
				 * 服务端的资源
				 */
                try {
                    System.out.println(clientHost + "下线了.");
					/*
					 * 将该客户端的输出流从共享集合中删除
					 */
                    if (writer != null) {
                        removeOut(writer);
                    }
                    //删除该客户端输出流后，输出在线人数
                    System.out.println("当前在线人数:" + allOut.size());

					/*
					 * 关闭该客户端的socket会自动将
					 * 对应该客户端的输入流与输出流
					 * 关闭了
					 */
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 该任务的作用是轮询消息队列，将其中的每一条数据
     * 转发给所有客户端
     * 使用这种方式可以保证所有客户端的输出流只被当前
     * 这一个线程所访问
     *
     * @author admin
     */
    private class SendMessageToAllClientHandler implements Runnable {
        public void run() {
            try {
                String message = null;
				/*
				 * 要循环获取消息队列中的信息
				 * 向消息队列要数据这件事情不能只看当前集合
				 * 是否还有元素。
				 * 所以:要不能停
				 */
                while (true) {
					/*
					 * 若队列里存在元素
					 */
                    if (messageQueue.size() > 0) {
						/*
						 * 取出队首元素，转发给所有客户端
						 */
                        message = messageQueue.poll();
                        sendMessageToAllClient(message);
                    } else {
						/*
						 * 若没有元素,就让当前线程休息一会
						 * 休息0.3秒，不能休眠时间过长，要
						 * 保证转发的即时感。
						 */
                        try {
                            Thread.sleep(300);
                        } catch (InterruptedException e) {
                        }
                    }

                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    public static void main(String[] args) {
        Server server = new Server();
        //启动服务端
        server.start();
    }
}





